package com.chukun.flink.stream.action.fraud.rules.utils;

import com.chukun.flink.stream.action.fraud.rules.accumulators.AverageAccumulator;
import com.chukun.flink.stream.action.fraud.rules.accumulators.BigDecimalCounter;
import com.chukun.flink.stream.action.fraud.rules.accumulators.BigDecimalMaximum;
import com.chukun.flink.stream.action.fraud.rules.accumulators.BigDecimalMinimum;
import com.chukun.flink.stream.action.fraud.rules.model.Rule;
import org.apache.flink.api.common.accumulators.SimpleAccumulator;

import java.math.BigDecimal;

/**
 * 规则工具
 * @author chukun
 */
public class RuleHelper {

  /**
   * 生成具体的聚合处理器
   * @param rule
   * @return
   */
  public static SimpleAccumulator<BigDecimal> getAggregator(Rule rule) {
    switch (rule.getAggregatorFunctionType()) {
      case SUM:
        return new BigDecimalCounter();
      case AVG:
        return new AverageAccumulator();
      case MAX:
        return new BigDecimalMaximum();
      case MIN:
        return new BigDecimalMinimum();
      default:
        throw new RuntimeException(
            "Unsupported aggregation function type: " + rule.getAggregatorFunctionType());
    }
  }
}